package org.nestor.demo.client;

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import org.nestor.demo.protobuf.Buyer;
import org.nestor.demo.protobuf.Order;
import org.nestor.demo.protobuf.OrderQueryClientStreamGrpc;
import org.nestor.demo.protobuf.OrderQueryGrpc;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.Iterator;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 测试客户端流式与双向流式
 *
 * @author : Nestor.Bian
 * @version : V 1.0
 * @date : 2025/3/6
 */
public class OrderClientStreamGRPCClient {
    private static final String host = "localhost";
    private static final int serverPort = 9999;

    public static void main(String[] args) {
        //1,拿到一个通信channel
        ManagedChannel channel = ManagedChannelBuilder.forAddress(host, serverPort).
                usePlaintext()// 这里将使用纯文本，无需任何加密
                .build();
        CountDownLatch latch = new CountDownLatch(1);
        try {
            //2.拿到stub对象
            final OrderQueryClientStreamGrpc.OrderQueryClientStreamStub stub = OrderQueryClientStreamGrpc.newStub(channel);
            // gRPC的请求参数
            DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            // 通过stub发起远程gRPC请求
            // gRPC的响应
            final StreamObserver<Buyer> buyerStreamObserver = stub.listOrders(new StreamObserver<Order>() {
                @Override
                public void onNext(Order order) {
                    System.out.println("订单ID：" + order.getOrderId() + "产品ID：" + order.getProductId() + "日期：" +
                            // 使用DateTimeFormatter将时间戳转为字符串
                            dtf.format(LocalDateTime.ofEpochSecond(order.getOrderTime(), 0, ZoneOffset.of("+8"))) + "备注：" + order.getBuyerRemark());
                }

                @Override
                public void onError(Throwable t) {
                    t.printStackTrace();
                    latch.countDown();
                }

                @Override
                public void onCompleted() {
                    System.out.println("response finished");
                    latch.countDown();
                }
            });

            for (int i = 0; i <2; i++) {
                Buyer buyer = Buyer.newBuilder().setBuyerId(100 + i).build();
                buyerStreamObserver.onNext(buyer);
            }
            buyerStreamObserver.onCompleted();

            // 等待直到通话结束
            latch.await(5, TimeUnit.SECONDS);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 5.关闭channel, 释放资源.
            channel.shutdown();
        }

    }

}
